package com.neuedu.grains

import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.IntegerType

object Grains {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME","root")
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
    val sparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport()
      .appName(name = "grains")
      .getOrCreate()
    val originPath = "/grains/data/grains.1734542341611.csv";
    val df = sparkSession.read.option("header","true").csv(originPath);

    val cols = Array("year","GrainProduction","CerealYield","RiceYield","WheatYield","CornYield","LegumesYield","PotatoProduction","Sown_area_of_agricultural_crops","Sown_area_of_grain_crops","Grain_sown_area","Paddy_sown_area","Wheat_sown_area","Corn_sown_area","Beans_are_sown_in_area","Potato_sown_area","Allpeople","Per_capita_grain")
    df.na.drop(cols)
    df.show()
    val table = df.select(
      col("year") cast(IntegerType),
      col("GrainProduction") cast(IntegerType),
      col("CerealYield") cast(IntegerType),
      col("RiceYield") cast(IntegerType),
      col("WheatYield") cast(IntegerType),
      col("CornYield") cast(IntegerType),
      col("LegumesYield") cast(IntegerType),
      col("PotatoProduction") cast(IntegerType),
      col("Sown_area_of_agricultural_crops") cast(IntegerType),
      col("Sown_area_of_grain_crops") cast(IntegerType),
      col("Grain_sown_area") cast(IntegerType),
      col("Paddy_sown_area") cast(IntegerType),
      col("Wheat_sown_area") cast(IntegerType),
      col("Corn_sown_area") cast(IntegerType),
      col("Beans_are_sown_in_area") cast(IntegerType),
      col("Potato_sown_area") cast(IntegerType),
      col("Allpeople") cast(IntegerType),
      col("Per_capita_grain") cast(IntegerType),
    )

    table.write.mode(SaveMode.Overwrite).saveAsTable("db_grains.all_grains")

    sparkSession.stop()
    sparkSession.close()
  }
}
